import json
from typing import Any, Optional, Callable

# Lazy import redis; fall back to None if unavailable
try:
    import redis  # type: ignore
except Exception:
    redis = None  # type: ignore

from ..config.redis_config import get_redis_settings


class _MemoryRedis:
    """Minimal in-memory Redis-like client for demo/testing fallback.
    Supports set/get/delete/exists/expire/ping used by RedisManager.
    """
    def __init__(self) -> None:
        self._store: dict[str, Any] = {}

    def ping(self) -> bool:
        return True

    def set(self, key: str, value: Any, ex: Optional[int] = None) -> None:
        self._store[key] = value

    def get(self, key: str) -> Optional[Any]:
        return self._store.get(key)

    def delete(self, key: str) -> int:
        return 1 if self._store.pop(key, None) is not None else 0

    def exists(self, key: str) -> bool:
        return key in self._store

    def expire(self, key: str, ttl_seconds: int) -> bool:
        # TTL unsupported in memory fallback; behave as no-op
        return True


class RedisManager:
    """
    Redis 管理器模块（单一职责：提供统一的 Redis 访问入口与 JSON/字符串读写辅助）。
    - 支持在缺少 redis 库时回退到内存实现（_MemoryRedis），保证测试与示例可运行
    - 对外暴露 REDIS_MANAGER 单例，以及便捷的模块级函数（redis_set/redis_get/...）
    - 键名约定：流程上下文写入使用 {runId}_context
    """

    def __init__(self) -> None:
        self._settings = get_redis_settings()
        self._client: Optional[Any] = None

    def _ensure_client(self) -> None:
        if not self._settings.enabled:
            raise RuntimeError("Redis is disabled via REDIS_ENABLED=false")
        if self._client is None:
            if redis is None:
                # Fallback to in-memory client when redis module is unavailable
                self._client = _MemoryRedis()
            else:
                self._client = redis.Redis(
                    host=self._settings.host,
                    port=self._settings.port,
                    password=self._settings.password,
                    db=self._settings.db,
                    ssl=self._settings.ssl,
                    decode_responses=True,  # return str for string operations
                    socket_timeout=5,        # short timeout for tests
                )

    def client(self) -> Any:
        self._ensure_client()
        assert self._client is not None
        return self._client

    # Convenience helpers
    def ping(self) -> bool:
        c = self.client()
        return c.ping()

    def set_str(self, key: str, value: str) -> None:
        """Backward-compatible: set plain string value."""
        c = self.client()
        c.set(key, value)

    def set_json(self, key: str, obj: Any) -> None:
        """Backward-compatible: store JSON-serialized object."""
        c = self.client()
        c.set(key, json.dumps(obj, ensure_ascii=False))

    def get_str(self, key: str) -> Optional[str]:
        """Backward-compatible: get plain string value (or None)."""
        c = self.client()
        v = c.get(key)
        return v if isinstance(v, str) else (v.decode("utf-8") if v is not None else None)

    def get_json(self, key: str) -> Optional[Any]:
        """Backward-compatible: get JSON and deserialize, return None if not JSON."""
        c = self.client()
        v = c.get(key)
        if v is None:
            return None
        if isinstance(v, bytes):
            v = v.decode("utf-8")
        try:
            return json.loads(v)
        except Exception:
            return None

    def delete(self, key: str) -> int:
        """Delete key, return number of keys removed."""
        c = self.client()
        return int(c.delete(key))

    def exists(self, key: str) -> bool:
        """Check if key exists."""
        c = self.client()
        return bool(c.exists(key))

    def get_or_set(self, key: str, factory: Callable[[], Any], ttl_seconds: Optional[int] = None) -> Any:
        """Get existing value, or compute via factory and set it atomically if absent."""
        current = self.get(key)
        if current is not None:
            return current
        value = factory()
        # Store computed value
        self.set(key, value, ttl_seconds=ttl_seconds)
        return value

    def get(self, key: str) -> Optional[Any]:
        """Flexible get: try JSON first, fallback to plain string."""
        obj = self.get_json(key)
        if obj is not None:
            return obj
        return self.get_str(key)
    
    def set(self, key: str, value: Any, ttl_seconds: Optional[int] = None) -> None:
        """Flexible set: dict/list as JSON, others as string. Supports TTL (seconds)."""
        if isinstance(value, (dict, list)):
            self.set_json(key, value)
            if ttl_seconds:
                self.client().expire(key, ttl_seconds)
        else:
            if ttl_seconds is not None:
                self.client().set(key, str(value), ex=ttl_seconds)
            else:
                self.set_str(key, str(value))


# Singleton instance
REDIS_MANAGER = RedisManager()


# Module-level simple wrappers for easier usage

def redis_set(key: str, value: Any, ttl_seconds: Optional[int] = None) -> None:
    """
    便捷写入：
    - dict/list 以 JSON 写入；其他类型写入为字符串
    - 支持可选 TTL（秒）
    """
    # Use flexible setter when available; fall back to specific methods
    if isinstance(value, (dict, list)):
        REDIS_MANAGER.set_json(key, value)
        if ttl_seconds:
            # Redis doesn't support TTL with set_json directly; apply with expire
            REDIS_MANAGER.client().expire(key, ttl_seconds)
    else:
        if ttl_seconds is not None:
            REDIS_MANAGER.client().set(key, str(value), ex=ttl_seconds)
        else:
            REDIS_MANAGER.set_str(key, str(value))


def redis_get(key: str) -> Optional[Any]:
    """
    便捷读取：优先尝试 JSON 反序列化，若失败则返回字符串（或 None）。
    """
    # Try JSON first, then plain string
    obj = REDIS_MANAGER.get_json(key)
    if obj is not None:
        return obj
    return REDIS_MANAGER.get_str(key)


def redis_delete(key: str) -> int:
    return REDIS_MANAGER.delete(key)


def redis_exists(key: str) -> bool:
    return REDIS_MANAGER.exists(key)


def redis_get_or_set(key: str, factory: Callable[[], Any], ttl_seconds: Optional[int] = None) -> Any:
    """
    若键存在则直接返回，否则通过 factory() 计算并写入，再返回值。
    可选设置 TTL（秒）。
    """
    return REDIS_MANAGER.get_or_set(key, factory, ttl_seconds=ttl_seconds)